package Spark._sql.day02

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Milliseconds, StreamingContext}


object SparkStreaming {

  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]): Unit = {
    //创建SparkstreamingContext
    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)

    val ssc: StreamingContext= new StreamingContext(conf, Milliseconds(10000))

    //创建DStream
    val res: ReceiverInputDStream[String] = ssc.socketTextStream("doit01",7777)

    //调用算子
    val res2: DStream[(String, Int)] = res.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

    res2.print()

    //启动Sparkstreaming
    ssc.start()


    //优雅跑路

ssc.awaitTermination()



  }





}
